package com.xinjian.jianba.service.impl;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.spatial4j.core.distance.GeodesicSphereDistCalc.Haversine;
import javax.annotation.PostConstruct;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.spatial4j.core.distance.DistanceUtils;
import com.xinjian.jianba.dto.es.EsServiceResult;
import com.xinjian.jianba.dto.es.FindDocDto;
import com.xinjian.jianba.es.pool.ElasticSearchPool;
import com.xinjian.jianba.exception.BusinessException;
import com.xinjian.jianba.exception.ResponseCode;
import com.xinjian.jianba.exception.ResponseCodeAwareException;
import com.xinjian.jianba.service.EsService;
import com.xinjian.jianba.utils.EsUtil;
import com.xinjian.jianba.utils.JianBaEnum;
import com.xinjian.jianba.utils.LocationUtils;


@Service
public class EsServiceImpl implements EsService{

	private static Logger LOGGER=LoggerFactory.getLogger(EsServiceImpl.class);
	private static EsServiceImpl esServiceImpl;
	@Autowired
	private ElasticSearchPool elasticSearchPool;
	@PostConstruct
	public void init() {
		esServiceImpl = this;
	}
	public static RestHighLevelClient getesConnection(){
		RestHighLevelClient EsClient = esServiceImpl.elasticSearchPool.getResource();
		return EsClient;
	}
	public static void closeEsConnection(RestHighLevelClient client){
		esServiceImpl.elasticSearchPool.returnResource(client);
	}
	@Override
	public EsServiceResult indexDocumentUserId(String index, String type, XContentBuilder builder,String docId) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		//创建索引请求
		IndexRequest request=null;
		if(docId==null||docId.equals("")){
		 request = new IndexRequest(
				index,   //索引
				type//类型 ,6x版本以后一个索引只有一个类型
				);
		}else{
		request = new IndexRequest(
					index,   //索引
					type,//类型 ,6x版本以后一个索引只有一个类型
					docId);//文档id
		}
		if(builder!=null){
			request.source(builder);
		}
		request.routing(docId);
		request.timeout(TimeValue.timeValueSeconds(20));
		request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
		request.opType(DocWriteRequest.OpType.CREATE);
        //4、发送请求
        IndexResponse indexResponse = null;
        try {
            // 同步方式
            indexResponse = client.index(request);  
            esServiceResult.setStatus(EsServiceResult.ResultStatus.CREATED.getValue());
            esServiceResult.setDocId(indexResponse.getId());
        } catch(ElasticsearchException e) {
            // 捕获，并处理异常
            //判断是否版本冲突、create但文档已存在冲突
            if (e.status() == RestStatus.CONFLICT) {
            	esServiceResult.setStatus(EsServiceResult.ResultStatus.CONFLICT.getValue()); 	
            }else{
            	LOGGER.error("操作失败indexDocumentUserId->索引异常", e);
            	esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
            }
        }catch(Exception e){
        	LOGGER.error("操作失败indexDocumentRandomId" , e.getMessage());
        	esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
        }finally{
        	closeEsConnection(client);
        }
        
        //5、处理响应
//        if(indexResponse != null) {
//            String indexre = indexResponse.getIndex();
//            String typere = indexResponse.getType();
//            String id = indexResponse.getId();
//            long version = indexResponse.getVersion();
//            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
//            	//LOGGER.info("新增文档成功，处理逻辑代码写到这里");
//            } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
//            	//LOGGER.info("修改文档成功，处理逻辑代码写到这里");
//            }
//            // 分片处理信息
//            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
//            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
//                
//            }
//            // 如果有分片副本失败，可以获得失败原因信息
//            if (shardInfo.getFailed() > 0) {
//                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
//                    String reason = failure.reason(); 
//                    System.out.println("副本失败原因：" + reason);
//                }
//            }
//        }
        
		return esServiceResult;
	}
	@Override
	public EsServiceResult updateDocumentsUserId(String index, String type, XContentBuilder builder, String docId) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		UpdateRequest request = new UpdateRequest(
				index, 
				type,  
				docId).doc(builder);
		request.timeout(TimeValue.timeValueSeconds(20));
		request.waitForActiveShards(ActiveShardCount.ALL);
		request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
		UpdateResponse updateResponse=null;
		try {
			 updateResponse = client.update(request);
			 esServiceResult.setStatus(EsServiceResult.ResultStatus.UPDATED.getValue());
		} catch (ElasticsearchException e) {
			if (e.status() == RestStatus.NOT_FOUND) {
				esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue());
			}else{
				LOGGER.error("操作失败 updateDocumentsUserId->索引异常", e);
				esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
			}
		}catch (Exception e) {
			LOGGER.error("操作失败EXCEPTION updateDocumentsUserId" , e.getMessage());
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		}finally{
			closeEsConnection(client);
		}
		return esServiceResult;
	}
	@Override
	public EsServiceResult upsrts(String index, String type, XContentBuilder builder, String docId) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		UpdateRequest request = new UpdateRequest(
				index, 
				type,  
				docId).doc(builder);
		request.routing(docId);
		request.timeout(TimeValue.timeValueSeconds(20));
		request.waitForActiveShards(ActiveShardCount.ALL);
		request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
		request.docAsUpsert(true);
		UpdateResponse updateResponse=null;
		try {
			 updateResponse = client.update(request);
			 if (updateResponse.getResult().getOp()==EsServiceResult.ResultStatus.CREATED.getValue()) {
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.CREATED.getValue());
			 } else if (updateResponse.getResult().getOp()==EsServiceResult.ResultStatus.UPDATED.getValue()) {
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.UPDATED.getValue());
			 } else {
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
			 }
		} catch (ElasticsearchException e) {
			LOGGER.error("操作失败upsrts->索引异常", e);
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		
		}catch (Exception e) {
			LOGGER.error("操作失败upsrts" , e.getMessage());
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		}finally{
			closeEsConnection(client);
		}
		return esServiceResult;
	}
	@Override
	public EsServiceResult findByDocId(String index, String type,String docId) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		GetRequest request = new GetRequest(
				index, 
				type,  
		        docId);
		request.routing(docId);
		GetResponse response=null;
		try {
			 response = client.get(request);
			 if(response.isExists()){
			 Map<String,Object> map=response.getSource();
			 esServiceResult.setMap(map);
			 esServiceResult.setDocId(response.getId());
			 esServiceResult.setStatus(EsServiceResult.ResultStatus.SUCCESS.getValue());
			 }else{
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue());
			 }
		} catch (ElasticsearchException e) {
			if (e.status() == RestStatus.NOT_FOUND) {
				esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue());
			}else{
				LOGGER.error("操作失败findByDocId->索引异常", e);
				esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
			}
		}catch(Exception e){
			LOGGER.error("操作失败findByDocId" , e.getMessage());
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		}finally {
			closeEsConnection(client);
		}
		
		return esServiceResult;
	}
	@Override
	public EsServiceResult findPersons(String index, String type,FindDocDto findDocDto) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
		//拼接请求body
		QueryBuilder sexQuery=null;
		if(findDocDto.getMeetType()==JianBaEnum.SexEnum.MALE.getValue()){
		  sexQuery=QueryBuilders.termQuery(EsUtil.sex, JianBaEnum.SexEnum.MALE.getValue());//男 0
		}else if(findDocDto.getMeetType()==JianBaEnum.SexEnum.FEMALE.getValue()){
		  sexQuery=QueryBuilders.termQuery(EsUtil.sex, JianBaEnum.SexEnum.FEMALE.getValue());//女1
		}else{
		  sexQuery=QueryBuilders.termsQuery(EsUtil.sex, new int[]{JianBaEnum.SexEnum.MALE.getValue(),JianBaEnum.SexEnum.FEMALE.getValue()});//男女
		}
		QueryBuilder isOnlineQuery=null;
		if(findDocDto.getIsOnline()==0){
			 	isOnlineQuery=QueryBuilders.termQuery(EsUtil.isOnline, 0);//下线 0
			}else if(findDocDto.getIsOnline()==1){
				isOnlineQuery=QueryBuilders.termQuery(EsUtil.isOnline, 1);//上线1
			}else{
				isOnlineQuery=QueryBuilders.termsQuery(EsUtil.isOnline, new int[]{0,1});//
			}
		
		 QueryBuilder sexAndOnlineAndTitleQuery=null;
		 if(findDocDto.getTheme()!=null&&!findDocDto.getTheme().equals("")){
			 QueryBuilder themeQuery=QueryBuilders.matchQuery(EsUtil.theme, findDocDto.getTheme());
		     sexAndOnlineAndTitleQuery = QueryBuilders.boolQuery().must(sexQuery).must(isOnlineQuery).must(themeQuery);
		 }else{
			 sexAndOnlineAndTitleQuery = QueryBuilders.boolQuery().must(sexQuery).must(isOnlineQuery);
		 }
		 //获取距离
		 
		 sourceBuilder.query(sexAndOnlineAndTitleQuery).from((findDocDto.getPageNumber()-1)*findDocDto.getPageSize()).size(findDocDto.getPageSize());
		 SearchRequest searchRequest = new SearchRequest(index)
	                .source(sourceBuilder.sort(SortBuilders.geoDistanceSort("location", findDocDto.getLatfloat(), findDocDto.getLngfloat())
	                        .order(SortOrder.ASC)
	                        .unit(DistanceUnit.METERS)));
//		 LOGGER.info("BODY--->{}",sourceBuilder.sort(SortBuilders.geoDistanceSort("location", findDocDto.getLatfloat(), findDocDto.getLngfloat())
//	                        .order(SortOrder.ASC)
//	                        .unit(DistanceUnit.KILOMETERS)));
		 SearchResponse searchResponse=null;
		//发送请求
		try {
			searchResponse = client.search(searchRequest);
			List<Map<String,Object>> list=new ArrayList<Map<String,Object>>();
			 for (SearchHit hit : searchResponse.getHits().getHits()) {  
	             Map<String, Object> map=hit.getSourceAsMap();
//	             List<Double> location=(List<Double>)map.get(EsUtil.location);
//	             double calDistance=LocationUtils.getDistance(findDocDto.getLatfloat(), findDocDto.getLngfloat(), location.get(1), location.get(0));
//	             double calDistance=DistanceUtils.distHaversineRAD(lat, lon, location.get(1), location.get(0));
	             if(String.valueOf(findDocDto.getUserId()).equals(map.get(EsUtil.userId)+"")){
	            	 continue;
	             }
	             Object[] calDistance=hit.getSortValues();
	             map.put("calDistance", calDistance[0]);
	             list.add(map);
	          } 
			 if(list.size()==0){
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue()); 
			 }else{
				 esServiceResult.setStatus(EsServiceResult.ResultStatus.SUCCESS.getValue());
				 esServiceResult.setList(list);
			 }
		} catch (ElasticsearchException e) {
			LOGGER.error("操作失败findPersons->索引异常", e);
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		}catch(Exception e){
			LOGGER.error("操作失败findPersons->索引异常", e);
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		}finally{
			closeEsConnection(client);
		}
		
		 
		 return esServiceResult;
	}
	@Override
	public EsServiceResult deleteDocByDocId(String index, String type, String docId) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		DeleteRequest request = new DeleteRequest(
				index,    
				type,     
				docId);  
		request.routing(docId);
		request.timeout(TimeValue.timeValueSeconds(20));
		request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
		try {
			DeleteResponse deleteResponse = client.delete(request);
			if (deleteResponse.getResult().getOp()== EsServiceResult.ResultStatus.NOT_FOUND.getValue()) {
				esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue());
			}else{
				esServiceResult.setStatus(EsServiceResult.ResultStatus.DELETED.getValue());
			}
		} catch (ElasticsearchException e) {
			if (e.status() == RestStatus.CONFLICT) {
				esServiceResult.setStatus(EsServiceResult.ResultStatus.CONFLICT.getValue());
				LOGGER.error("操作失败deleteDocByDocId->索引异常", e);
		    }else{
		    	esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
		    	LOGGER.error("操作失败deleteDocByDocId->索引异常", e);
		    }
		}catch(Exception e){
			esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
	    	LOGGER.error("操作失败deleteDocByDocId->索引异常", e);
		}finally {
			closeEsConnection(client);
		}
		return esServiceResult;
	}
	/* 
	 * 通用查询方法,(前提)需要匹配距离时,es中距离字段必须命名为 "location"
	 */
	@Override
	public EsServiceResult find(String index, String type, QueryBuilder queryBuilder,Float lngfloat,Float latfloat,Integer pageNumber,Integer pageSize) {
		EsServiceResult esServiceResult=new EsServiceResult();
		RestHighLevelClient client=getesConnection();
		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
		int status=0;
		if(queryBuilder!=null){
			sourceBuilder=sourceBuilder.query(queryBuilder);
			status=1;
		}
		if(latfloat!=null&&lngfloat!=null){
			sourceBuilder=sourceBuilder.sort(SortBuilders.geoDistanceSort("location", latfloat, lngfloat)
                    .order(SortOrder.ASC)
                    .unit(DistanceUnit.METERS));
			status=1;
		}
		if(pageNumber!=null&&pageSize!=null){
			sourceBuilder=sourceBuilder.from((pageNumber-1)*pageSize).size(pageSize);
			status=1;
		}
		if(status==0){
			throw new BusinessException(ResponseCode.ERROR, "查询条件为空");
		}
		SearchRequest searchRequest = new SearchRequest(index).source(sourceBuilder);
		 SearchResponse searchResponse=null;
			//发送请求
			try {
				searchResponse = client.search(searchRequest);
				List<Map<String,Object>> list=new ArrayList<Map<String,Object>>();
				 for (SearchHit hit : searchResponse.getHits().getHits()) {  
		             Map<String, Object> map=hit.getSourceAsMap();
		             Object[] calDistance=hit.getSortValues();
		             map.put("calDistance", calDistance[0]);
		             list.add(map);
		          } 
				 if(list.size()==0){
					 esServiceResult.setStatus(EsServiceResult.ResultStatus.NOT_FOUND.getValue()); 
				 }else{
					 esServiceResult.setStatus(EsServiceResult.ResultStatus.SUCCESS.getValue());
					 esServiceResult.setList(list);
				 }
			} catch (ElasticsearchException e) {
				LOGGER.error("操作失败findPersons->索引异常", e);
				esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
			}catch(Exception e){
				LOGGER.error("操作失败findPersons->索引异常", e);
				esServiceResult.setStatus(EsServiceResult.ResultStatus.ERROR.getValue());
			}finally{
				closeEsConnection(client);
			}
			 return esServiceResult;
	}
	

}
